package org.box.demos.multithreading.wgetmt.j16mt;

import java.rmi.UnexpectedException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import javax.xml.ws.http.HTTPException;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.protocol.BasicHttpContext;
import org.apache.http.protocol.HttpContext;
import org.apache.http.util.EntityUtils;
import org.box.demos.multithreading.wgetmt.core.ContentType;
import org.box.demos.multithreading.wgetmt.core.URL;

/**
 * TaskManager controls over {@link DownloadTask}, downloads requested urls and
 * fires events on download completion or failure to {@link TaskManager}
 * listeners in the main thread (no synchronization is required)
 * 
 * The class is thread safe.
 * 
 * @author <a href="mailto:alexander.box@gmail.com">Alexander Korobka</a>
 * 
 */
public class TaskManager {

	private static final long DEFAULT_THREAD_TIMEOUT = TimeUnit.MILLISECONDS
			.convert(1L, TimeUnit.MINUTES);

	private final List<TaskResult> completedTasks = new CopyOnWriteArrayList<TaskResult>();

	private final List<URLInfo> requestedUrls = new LinkedList<URLInfo>();
	private final List<TaskInfo> activeTasks = new LinkedList<TaskInfo>();
	private volatile int remainsNumber;

	private final ExecutorService pool;
	private final HttpClient httpClient;
	private final Object threadMonitor = new Object();

	private long threadTimeout = DEFAULT_THREAD_TIMEOUT;
	private final int nThreads;

	private final List<TaskManagerListener> listeners = new CopyOnWriteArrayList<TaskManagerListener>();

	/**
	 * The default constructor
	 * 
	 * @param httpClient
	 *            - an instance of the prepared HttpClient
	 * @param nThreads
	 *            - the number of thread to perform
	 */
	public TaskManager(HttpClient httpClient, int nThreads) {
		this.httpClient = httpClient;
		this.nThreads = nThreads;
		this.pool = Executors.newCachedThreadPool();
	}

	/**
	 * Adds new urls to download list, if it's not download already
	 * 
	 * @param groupId
	 *            - urls group identifier or url identifier, if single url .
	 * @param urls
	 *            - a list of urls to download
	 * 
	 */
	public void requestUrls(String groupId, Collection<URL> urls) {
		synchronized (requestedUrls) {
			for (URL url : urls) {
				requestedUrls.add(new URLInfo(url, groupId));
			}
			remainsNumber = requestedUrls.size();
		}

	}

	/**
	 * Starts download process and waits completion of all download tasks
	 */
	public void runAndWait() {

		for (;;) {

			// clean up of all completed tasks from the active task list
			if (activeTasks.size() > 0) {
				List<TaskInfo> deleteList = new ArrayList<TaskInfo>();
				for (TaskInfo taskInfo : activeTasks) {
					if (taskInfo.getFuture().isDone()) {
						try {
							taskInfo.getFuture().get(10L, TimeUnit.SECONDS);
						} catch (Throwable e) {
							doTaskFailed(taskInfo, taskInfo.getTask()
									.getUrlInfo(), e);
						}
						deleteList.add(taskInfo);
					} else {
						// terminate hanged up tasks
						if (System.currentTimeMillis()
								- taskInfo.getStartTime() > getThreadTimeout()) {

							doTaskFailed(taskInfo, taskInfo.getTask()
									.getUrlInfo(), new TimeoutException());
							deleteList.add(taskInfo);
						}
					}
				}

				activeTasks.removeAll(deleteList);
			}

			// do complete all finished tasks
			while (completedTasks.size() != 0) {
				TaskResult result = completedTasks.remove(0);
				if (result.getContent() != null) {
					doTaskCompleted(result);
				} else {
					doTaskSkipped(result);
				}
			}

			try {
				synchronized (threadMonitor) {
					// create new tasks for requested Urls. The number of active tasks
					// have not to be higher then nThreads
					if (activeTasks.size() < nThreads) {
						List<URLInfo> lockedUrls = new ArrayList<URLInfo>(nThreads);
						synchronized (requestedUrls) {
							for (int i = 0; i < nThreads - activeTasks.size(); i++) {
								if (requestedUrls.size() > 0) {
									lockedUrls.add(requestedUrls.remove(0));
								}
							}
						}
		
						for (URLInfo url : lockedUrls) {
							TaskInfo taskInfo = new TaskInfo(new DownloadTask(url),
									System.currentTimeMillis());
		
							activeTasks.add(taskInfo);
							taskInfo.setFuture(pool.submit(taskInfo.getTask()));
						}
					}
		
					// exit if no active tasks and requested urls
					synchronized (requestedUrls) {
						if (activeTasks.size() == 0 && requestedUrls.size() == 0) {
							pool.shutdown();
							return;
						}
					}

					threadMonitor.wait(getThreadTimeout());
				}

			} catch (InterruptedException e) {
				// no response from all active threads
				// all hanged up tasks will be canceled in the next loop
				// no exception handlers are required
				// most likely the problem due to httpClient so let's re-init it
				httpClient.getConnectionManager().closeExpiredConnections();
			}
		}
	}
	

	public void truncateQueue() {
		synchronized (requestedUrls) {
			requestedUrls.clear();
		}
	}

	/**
	 * Performs removing url from locked list, adds into canceled list and
	 * notifies listeners about task's completion.
	 * 
	 * @param taskInfo
	 *            - TaskResult class to pass for listeners.
	 * @param url
	 *            - the given url
	 * @param e
	 */
	protected void doTaskFailed(TaskInfo taskInfo, URLInfo url, Throwable e) {
		remainsNumber--;
		taskInfo.getFuture().cancel(true);
		notifyUrlFailed(url, e);
	}

	/**
	 * Performs removing url from locked list, adds into completed list and
	 * notifies listeners about task's completion.
	 * 
	 * @param result
	 *            - TaskResult class to pass for listeners.
	 */
	protected void doTaskSkipped(TaskResult result) {
		remainsNumber--;
		notifyUrlSkipped(result);
	}

	/**
	 * Performs removing url from locked list, adds into completed list and
	 * notifies listeners about task's completion.
	 * 
	 * @param result
	 *            - TaskResult class to pass for listeners.
	 */
	protected void doTaskCompleted(TaskResult result) {
		remainsNumber--;
		notifyUrlDownloaded(result);
	}

	/**
	 * Fires an event if Url downloading has been skipped.
	 * 
	 * @param result
	 *            - TaskResult class to pass for listeners.
	 */
	protected void notifyUrlSkipped(TaskResult result) {
		for (TaskManagerListener listener : listeners) {
			listener.onTaskSkipped(result, remainsNumber);
		}
	}

	/**
	 * Fires an event on completion of Url download task.
	 * 
	 * @param result
	 *            - TaskResult class to pass for listeners.
	 */
	protected void notifyUrlDownloaded(TaskResult result) {
		for (TaskManagerListener listener : listeners) {
			listener.onTaskCompleted(result, remainsNumber);
		}
	}

	/**
	 * Fires an event on canceling of Url download task.
	 * 
	 * @param e
	 *            - TaskResult class to pass for listeners.
	 */
	protected void notifyUrlFailed(URLInfo id, Throwable e) {
		for (TaskManagerListener listener : listeners) {
			listener.onTaskFailed(id.getGroupId(), id.getUrl(), e);
		}
	}

	/**
	 * Adds a task's listener
	 * 
	 * @param listener
	 */
	public void addTaskListener(TaskManagerListener listener) {
		listeners.add(listener);
	}

	/**
	 * Getter for Thread Timeout field
	 * 
	 * @return a number of milliseconds to wait before a thread will be
	 *         terminated
	 */
	public long getThreadTimeout() {
		return threadTimeout;
	}

	/**
	 * Setter for Thread Timeout field
	 * 
	 * @param threadTimeout
	 *            - a number of milliseconds to wait before a thread will be
	 *            terminated
	 */
	public void setThreadTimeout(long threadTimeout) {
		this.threadTimeout = threadTimeout;
	}

	/**
	 * Represents a callable task for downloading a html-page by the given URL.
	 * 
	 * This class is inner due to it works with outter's completedTasks and
	 * threadMonitor private objects
	 * 
	 */
	private final class DownloadTask implements Callable<Void> {

		private final URLInfo urlInfo;

		/**
		 * The default constructor
		 * 
		 * @param urlInfo
		 *            - the given url unit to process
		 */
		private DownloadTask(URLInfo urlInfo) {
			this.urlInfo = urlInfo;
		}

		/**
		 * Returns the given url unit to download
		 * 
		 * @return string
		 */
		public URLInfo getUrlInfo() {
			return urlInfo;
		}

		/**
		 * {@link Callable}'s method call implementation. It uses Apache's Http
		 * Client from {@link TaskManager} to receiving data.
		 * 
		 */
		@Override
		public Void call() throws Exception {
			try {
				URL url = urlInfo.getUrl();
				HttpGet httpGet = new HttpGet(url.toString());
				HttpContext localContext = new BasicHttpContext();
				HttpResponse res = httpClient.execute(httpGet, localContext);
				
				if (res.getStatusLine().getStatusCode() != 200) {
					throw new HTTPException(res.getStatusLine().getStatusCode());
				}

				// Get hold of the response entity
				HttpEntity entity = res.getEntity();
				if (entity == null) {
					throw new UnexpectedException(
							"Unable to obtain a url's content: "
									+ url);
				}

				ContentType contentType = ContentType.parse(
						EntityUtils.getContentMimeType(entity));
				url.setContentType(contentType);
				
				byte[] content = EntityUtils.toByteArray(entity);

				// add new task result. Since we have to notify the main thread
				// about the task has been completed we mast use
				// threadMonitor.notifyAll() method that requires results
				// registration before notification and that is why we can't
				// use Future.get() to obtain results

				completedTasks.add(new TaskResult(
						urlInfo.getGroupId(), 
						urlInfo.getUrl(), 
						res.getStatusLine().getStatusCode(),
						content));
			} finally {
				// notify main thread about the task has been completed
				synchronized (threadMonitor) {
					threadMonitor.notifyAll();
				}
			}
			return null;
		}
	}

	/**
	 * Represent information about active tasks in the {@link ExecutorService}.
	 * It contains information about task's start time and {@link TaskResult} in
	 * the {@link Future} object.
	 * 
	 * There is no need to make this class thread safe since it uses in the main
	 * thread only.
	 * 
	 * @author box
	 * 
	 */
	private final class TaskInfo {

		private final DownloadTask downloadTask;
		private final long startTime;
		private Future<Void> future;

		/**
		 * Default constructor
		 * 
		 * @param downloadTask
		 * @param startTime
		 */
		public TaskInfo(DownloadTask downloadTask, long startTime) {
			this.downloadTask = downloadTask;
			this.startTime = startTime;
		}

		/**
		 * Setter for a future object
		 * 
		 * @param future
		 *            - object to set
		 */
		public void setFuture(Future<Void> future) {
			this.future = future;
		}

		/**
		 * Getter for a future object
		 * 
		 * @return future object for the given task
		 */
		public Future<Void> getFuture() {
			return future;
		}

		/**
		 * Getter for a {@link DownloadTask} object
		 * 
		 * @return an instance of {@link DownloadTask} object
		 */
		public DownloadTask getTask() {
			return downloadTask;
		}

		/**
		 * Returns a start time of the given task in milliseconds.
		 * 
		 * @return start time in milliseconds
		 */
		public long getStartTime() {
			return startTime;
		}

	}

	/**
	 * Serves as a descriptor of the requested url. Immutable class.
	 * 
	 * @author box
	 * 
	 */
	private final class URLInfo {

		private final URL url;
		private final String groupId;

		public URLInfo(URL url, String groupId) {
			this.url = url;
			this.groupId = groupId;
		}

		public String getGroupId() {
			return groupId;
		}

		public URL getUrl() {
			return url;
		}
	}
}
